package com.pajk.user.service.common.metaq;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.pajk.user.util.HessianUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.Serializable;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;

/**
 * User: haihua.chenhh
 * Date: 14-3-20
 * Time: 上午11:06
 */
public abstract class BaseConsumer implements MessageListenerConcurrently {

    private final static Logger logger = LoggerFactory.getLogger(BaseConsumer.class);
    protected DefaultMQPushConsumer consumer;
    protected String nameServer;
    protected int minConsumeThread = 2;
    protected int maxConsumeThread = 5;
    protected String group;
    protected String subExpression;
    protected TopicEnum topicEnum;
    // 定时消息相关
    // 线上环境：messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 40m 50m 1h 2h 6h
    // 开发环境：messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 40m 50m 1h 2h 6h 12h 1d
    private final static int[] DELAY_LEVELS = new int[]{3, 5, 9, 14, 15, 16, 17, 18, 19, 20, 21};
    protected int maxRetryCount = 10;

    public void init() throws MQClientException {
        if ("localTest".equals(nameServer)) {
            return;
        }
        if (StringUtils.isBlank(System.getProperty("rocketmq.namesrv.domain"))) {
            System.setProperty("rocketmq.namesrv.domain", nameServer);
        }
        topicEnum = getTopicEnum();
        if (StringUtils.isBlank(group)) {
            group = "S_" + topicEnum.getTopic() + "_" + topicEnum.getTags();
        }
        consumer = new DefaultMQPushConsumer(group);
        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumeThreadMin(minConsumeThread);
        consumer.setConsumeThreadMax(maxConsumeThread);
        //可以不设置 设置后可以起多个 消费端
        try {
            consumer.setInstanceName("DEFAULT_CONSUMER-"+ InetAddress.getLocalHost().getHostName());
        } catch (UnknownHostException e) {
            logger.error("getHostName error",e);
        }
        //设置订阅的topic 设置订阅过滤表达式
        if (StringUtils.isBlank(subExpression)) {
            subExpression = topicEnum.getTags();
            consumer.subscribe(topicEnum.getTopic(), subExpression);
        } else {
            consumer.subscribe(topicEnum.getTopic(), subExpression);
        }
        try {
            consumer.registerMessageListener(this);
            consumer.start();
        } catch (MQClientException e) {
            logger.error("consumer start error!topic={},subExpression={},group={}",
                    topicEnum.getTopic(), subExpression, group, e);
        }
        logger.info("consumer start! topic={},subExpression={},group={}", topicEnum.getTopic(), subExpression, group);
    }

    public void destroy() {
        if (consumer != null) {
            consumer.shutdown();
            logger.info("consumer shutdown! topic={0},subExpression={1},group={2}", topicEnum.getTopic(), subExpression, group);
        }
    }

    public abstract TopicEnum getTopicEnum();

    /**
     * 子类实现的，用于实际记录错误的代码
     *
     * @param msgObj 消息对象列表
     * @return 消费状态
     */
    public abstract void doLogErrorConsumeMessage(MsgObj msgObj);

    /**
     * 子类实现的，用于实际消费的代码
     *
     * @param msgObj 消息对象列表
     * @return 消费状态
     */
    public abstract ConsumeConcurrentlyStatus doConsumeMessage(MsgObj msgObj);

    /**
     * 基类实现消息监听接口，加上打印metaq监控日志的方法
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                    ConsumeConcurrentlyContext context) {
        long startTime = System.currentTimeMillis();
        logger.info("receive_message:{0}", msgs.toString());
        if (msgs == null || msgs.size() < 1) {
            logger.error("receive empty msg!");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

        List<Serializable> msgList = new ArrayList<Serializable>();
        for (MessageExt message : msgs) {
            msgList.add(decodeMsg(message));
        }

        final int reconsumeTimes = msgs.get(0).getReconsumeTimes();
        MsgObj msgObj=new MsgObj();
        msgObj.setReconsumeTimes(reconsumeTimes);
        msgObj.setMsgList(msgList);
        msgObj.setContext(context);
        context.setDelayLevelWhenNextConsume(getDelayLevelWhenNextConsume(reconsumeTimes));

        ConsumeConcurrentlyStatus status = doConsumeMessage(msgObj);
        logger.info("ConsumeConcurrentlyStatus:{}|cost:{}", status, System.currentTimeMillis() - startTime);
        return status;


    }

    /**
     * 根据重试次数设置重新消费延迟时间
     * 1s 10s 30s 2m 10m 30m 1h 2h 12h 1d
     *
     * @param reconsumeTimes 重试的次数
     * @return level级别
     */
    public int getDelayLevelWhenNextConsume(int reconsumeTimes) {
        if (reconsumeTimes >= DELAY_LEVELS.length) {
            return DELAY_LEVELS[DELAY_LEVELS.length - 1];
        }
        return DELAY_LEVELS[reconsumeTimes];
    }

    private Serializable decodeMsg(MessageExt msg) {
        if (msg == null) {
            return null;
        }
        //1.反序列化
        try {
            return HessianUtils.decode(msg.getBody());
        } catch (IOException e) {
            logger.error("反序列化出错!" + e.getMessage(), e);
            return null;
        }
    }


    public void setNameServer(String nameServer) {
        this.nameServer = nameServer;
    }

    public void setMinConsumeThread(int minConsumeThread) {
        this.minConsumeThread = minConsumeThread;
    }

    public void setMaxConsumeThread(int maxConsumeThread) {
        this.maxConsumeThread = maxConsumeThread;
    }

    public void setGroup(String group) {
        this.group = group;
    }

    public void setSubExpression(String subExpression) {
        this.subExpression = subExpression;
    }
}
